package com.mattgarner.jaddas.node;


import java.util.HashMap;
import java.util.Random;

import com.mattgarner.jaddas.common.ResultSet;
import com.mattgarner.jaddas.dataset.DataSet;
import com.mattgarner.jaddas.dataset.DataSetPartition;
import com.mattgarner.jaddas.node.data.ResultPair;

import com.sleepycat.je.Cursor;
import com.sleepycat.je.Database;
import com.sleepycat.je.DatabaseEntry;
import com.sleepycat.je.DatabaseException;
import com.sleepycat.je.LockMode;
import com.sleepycat.je.OperationStatus;

public class DataSetManager {
	private static DataSetManager instance;
	
	private HashMap<Integer, DataSet> onlineDataSets;
	private HashMap<String, Integer> dataSetNameToID;
	
	private NodeConfigurationManager nodeConfig;
	private ErrorLogManager logManager;
	private LocalDataProvider dataProvider;
	
	private Database dbDataSets;
	private Database dbDataSetPartitions;
	
	public static char dsPartitionKeyDelimeter = (char)31; 
	
	public DataSetManager() {
		onlineDataSets = new HashMap<Integer, DataSet>();
		dataSetNameToID = new HashMap<String, Integer>();
		nodeConfig = NodeConfigurationManager.getInstance();
		logManager = ErrorLogManager.getInstance();
		dataProvider = LocalDataProvider.getInstance();
		dbDataSets = dataProvider.openDatabase(nodeConfig.getSystemDataStore(), "SysDataSets", true);
		dbDataSetPartitions = dataProvider.openDatabase(nodeConfig.getSystemDataStore(), "SysDataSetPartitions", true);
		
		
		loadDataSetsFromDisk();
		loadDataSetPartitionsFromDisk();
		
	}
	
	// singleton method
	public final static DataSetManager getInstance() {
		if(instance == null) { 
			instance = new DataSetManager();
		}
		return instance;
	}
	
	
	// use synchronized methods only
	public synchronized final DataSet getDataSetByID(int dsID) {
		return onlineDataSets.get(dsID);
	}
	
	public synchronized final DataSet getDataSetByName(String dsName) {
		int dsID = 0;
		try {
			dsID = dataSetNameToID.get(dsName);
		} catch(RuntimeException e) {
			return null;
		}
		return onlineDataSets.get(dsID);
	}
	
	public final synchronized void addDataSet(DataSet dataSet) {
		addDataSet(dataSet, false);
	}
	
	public final synchronized void addDataSet(DataSet dataSet, boolean noWriteToDisk) {
		onlineDataSets.put(dataSet.getID(), dataSet);
		dataSetNameToID.put(dataSet.getName(), dataSet.getID());
		if(!noWriteToDisk) {
			try { 
				dataProvider.put(dbDataSets, dataSet.getID(), dataSet.serializeToByteArray());
				dataProvider.syncToDisk(dbDataSets);
			} catch(DatabaseException e) {
				dataProvider.writeToSysLog(dbDataSets, e.getMessage());
			}
		}
	}
	
	public final synchronized DataSet[] getOnlineDataSets() {
		return onlineDataSets.values().toArray(new DataSet[0]);
	}
	
	// generates a random unique ID in an attempt to minimize ID collisions
	public final synchronized int getNextDataSetID() {
		Random rand = new Random();
		int newID = Math.abs(rand.nextInt());
		Integer[] allIDs = onlineDataSets.keySet().toArray(new Integer[0]);
		for(int a=0; a<allIDs.length;a++) {
			if(allIDs[a] > newID) {newID = allIDs[a]; }
		}
		return newID+1;
	}
	
	public final synchronized void removeDataSet(DataSet dataSet) {
		onlineDataSets.remove(dataSet.getID());
		dataSetNameToID.remove(dataSet.getID());
		try { 
			dataProvider.remove(dbDataSets, dataSet.getID());
			dataProvider.syncToDisk(dbDataSets);
		} catch(DatabaseException e) {
			dataProvider.writeToSysLog(dbDataSets, e.getMessage());
		}
		
	}
	
	public final synchronized void loadDataSetsFromDisk() {
		try {
			ResultPair[] savedData = dataProvider.getAllValues(dbDataSets);
			for(int a =0; a<savedData.length; a++) {
				DataSet dSet = new DataSet(savedData[a].getValue());
				addDataSet(dSet, true);
				logManager.writeToLog(4, "DATASET_MGR", "Successfully loaded '"+dSet.getName()+"'");
			}
		} catch(DatabaseException e) {
			dataProvider.writeToSysLog(dbDataSets, e.getMessage());
		}
	}
	

	
	public final boolean checkCompatability(DataSet dataSet, ResultSet resultSet) {
		String dsTHCols = dataSet.getDataTableHeader().getColumnTypesString();
		String rdCols = resultSet.getHeader().getColumnTypesString();
		
		if(dsTHCols.matches(rdCols)) {
			return true;
		} else {
			return false;
		}
	   
	}
	
	public final void loadDataSetPartitionsFromDisk() {
		Cursor cursor = null;
		try {
			cursor = dbDataSetPartitions.openCursor(null, null);
			DatabaseEntry key = new DatabaseEntry();
			DatabaseEntry data = new DatabaseEntry();
		
			while(cursor.getNext(key, data, LockMode.DEFAULT) == OperationStatus.SUCCESS) {
				String keyValue = new String(key.getData(), "UTF-8");
				String[] keyValueParts = keyValue.split(""+dsPartitionKeyDelimeter);

				DataSet parentDataSet = this.getDataSetByID(Integer.parseInt(keyValueParts[0]));

				if(parentDataSet != null) {
					DataSetPartition dsPartition = new DataSetPartition(parentDataSet, data.getData());
					
					
					// get current tuple id
					String dsPartTupleIDKey = keyValue + dsPartitionKeyDelimeter + "tuple";
					String tupleString = dataProvider.getString(dbDataSetPartitions, dsPartTupleIDKey, LockMode.READ_COMMITTED);
					 
					dsPartition.setCurrentTupleID(Integer.parseInt(tupleString));
					
					parentDataSet.registerPartition(dsPartition);
					
					logManager.writeToLog(4, "DATASET_MGR", "Successfully registered partition '"+parentDataSet.getName()+":"+dsPartition.getPartitionID()+"'");
				}
			}
		} catch (Exception e) {
			logManager.writeToLog(1, "DATASET_MGR", "loadDataSetPartitionsFromDisk: Error loading partitions: " + e.getMessage());
		} finally {
			try { 
				cursor.close();
			} catch (Exception e) {
				// nothing
			}
		}
	
	}
	
	public final void saveDataSetPartitionToDisk(final DataSetPartition dsPartition) {
		try {
			String dsPartKey = "" + dsPartition.getParentDataSet().getID() + dsPartitionKeyDelimeter +  dsPartition.getPartitionID() ;
			byte[] dsPartBytes = dsPartition.serializeToByteArray();
			dataProvider.put(dbDataSetPartitions, dsPartKey, dsPartBytes);
			logManager.writeToLog(5, "DATASET_MGR", "saveDataSetPartitionToDisk: Stored partition ["+dsPartition.getPartitionID()+"].");
		} catch (Exception e) {
			logManager.writeToLog(1, "DATASET_MGR", "saveDataSetPartitionToDisk: Error saving partition ["+dsPartition.getPartitionID()+"]: " + e.getMessage());
		}	
	}
	
	public final void saveDataSetPartitionTupleIDToDisk(final DataSetPartition dsPartition) {
		try {
			String dsPartTupleIDKey = dsPartition.getParentDataSet().getID() + dsPartitionKeyDelimeter +  dsPartition.getPartitionID() + dsPartitionKeyDelimeter + "tuple";
			dataProvider.put(dbDataSetPartitions, dsPartTupleIDKey, new Integer(dsPartition.getCurrentTupleID()).toString());
		} catch (Exception e) {
			logManager.writeToLog(1, "DATASET_MGR", "saveDataSetPartitionTupleIDToDisk: Error saving partition ["+dsPartition.getPartitionID()+"]: " + e.getMessage());
		}	
	}
	
}


